package com.acme.domain.io;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.StreamCorruptedException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import com.acme.domain.DispatcherCommand;
import com.acme.domain.Stock;
import com.acme.domain.request.Request;
import com.acme.domain.usermap.UsersResponse;

public class Network implements Runnable {

    private final int port;
	private static ExecutorService dispatcher;
	private InetAddress addr;
	private Selector selector;
	private static Logger networkLogger = Logger.getLogger(Network.class.getName());
	private static Thread myThread;
	public Network(int port) {
		this.port = port;
		dispatcher = Executors.newSingleThreadExecutor();
		myThread = new Thread(this);
		myThread.start();
		networkLogger.info("Server started on port " + port);
	}

	@Override
	public void run() {
		try {
			this.selector = Selector.open();
			ServerSocketChannel serverChannel = ServerSocketChannel.open();
			serverChannel.configureBlocking(false);
					
			InetSocketAddress listenAddr = new InetSocketAddress(this.addr, this.port);
			serverChannel.socket().bind(listenAddr);
			serverChannel.register(this.selector, SelectionKey.OP_ACCEPT);
					
			while (!Thread.currentThread().isInterrupted()) {
				this.selector.select();
				
				Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
				while (keys.hasNext()) {
					SelectionKey key = (SelectionKey) keys.next();
					keys.remove();
					
					if (! key.isValid()) {
						continue;
					}
					
					if (key.isAcceptable() && key!=null) {
						this.accept(key);
					}
					else if (key.isReadable()) {
						this.read(key);
					}
					else if (key.isWritable()) {
						this.write(key);
					}
				}
			}
			
		} catch (IOException e) {
			networkLogger.warning(e.toString());
			e.printStackTrace();
		}
	}

	private void write(SelectionKey key) throws IOException {
		SocketChannel channel = (SocketChannel) key.channel();
		
		Object responseObject = UsersResponse.pollResponse(channel);
		if (responseObject != null) {
			ByteArrayOutputStream bos = new ByteArrayOutputStream();
			ObjectOutputStream oos = new ObjectOutputStream(bos);
			oos.writeObject(responseObject);
			oos.flush();
			channel.write(ByteBuffer.wrap(bos.toByteArray()));
		}
	}

	private void read(SelectionKey key) throws IOException {
		SocketChannel channel = (SocketChannel) key.channel();
		Socket socket = channel.socket();
		SocketAddress remoteAddr = socket.getRemoteSocketAddress();
		ByteBuffer buffer = ByteBuffer.allocate(2048);
		int numRead = -1;
		try {
			numRead = channel.read(buffer);
		} catch (IOException e){
		}
		
        if (numRead == -1) {
            networkLogger.info("Connection closed by client: " + remoteAddr);
            channel.close();
            key.cancel();
            return;
        }
		
	    InputStream is = new ByteArrayInputStream(buffer.array(), 0, buffer.limit());
		ObjectInputStream ois = new ObjectInputStream(is);

		try {
			Object o = ois.readObject();
			if (o instanceof Request ) {
				Request recievedRequest = (Request) o;
				dispatcher.execute( new DispatcherCommand(recievedRequest) );
			} else {
				if ( "w".equals(o) ){
					networkLogger.info("New user connected from " + remoteAddr);
					key.interestOps(SelectionKey.OP_READ);
				} else {
					UsersResponse.createUser(((String) o).substring(2), channel );
					key.interestOps(SelectionKey.OP_WRITE);
				}
			}
		} catch (StreamCorruptedException ex){
		} catch (ClassNotFoundException e) {
			networkLogger.warning( e.toString());
		}
	}

	private void accept(SelectionKey key) throws IOException {
		ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
		SocketChannel channel = serverChannel.accept();
		channel.configureBlocking( false );
		channel.register( selector, SelectionKey.OP_READ );
	}
	
	public static void shutdown(){
		for (Stock stock : DispatcherCommand.getStockMap().values()) {
			stock.shutdown();
		}
		dispatcher.shutdown();
		myThread.interrupt();
	}
}

